package com.test.rabbitmq.rpctest;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeoutException;

/**
 * Created with IntelliJ IDEA.
 * Description:
 *
 * @author lzy
 * @create 2020-05-24 15:05
 */
public class RPCClient implements AutoCloseable {
    private final Connection connection;
    private final Channel channel;
    private final String requestQueueName = "rpc_queue";

    public RPCClient() throws IOException, TimeoutException {
        // 1.We establish a connection and channel.
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        connection = factory.newConnection();
        channel = connection.createChannel();
    }

    public static void main(String[] argv) {
        try (RPCClient fibonacciRpc = new RPCClient()) {
            for (int i = 0; i < 32; i++) {
                String i_str = Integer.toString(i);
                System.out.println(" [x] Requesting fib(" + i_str + ")");
                String response = fibonacciRpc.call(i_str);
                System.out.println(" [.] Got '" + response + "'");
            }
        } catch (IOException | TimeoutException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 2. Our call method makes the actual RPC request.
     *
     * @param message
     * @return
     * @throws IOException
     * @throws InterruptedException
     */
    public String call(String message) throws IOException, InterruptedException {
        // 2.1 we first generate a unique correlationId number and save it -
        // our consumer callback will use this value to match the appropriate response.
        final String corrId = UUID.randomUUID().toString();

        // 2.2 Then, we create a dedicated exclusive queue for the reply and subscribe to it.
        String replyQueueName = channel.queueDeclare().getQueue();
        // 2.3 Next, we publish the request message, with two properties: replyTo and correlationId.
        AMQP.BasicProperties props = new AMQP.BasicProperties
                .Builder()
                .correlationId(corrId)
                .replyTo(replyQueueName)
                .build();

        channel.basicPublish("", requestQueueName, props, message.getBytes(StandardCharsets.UTF_8));

        final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);

        // 2.4 At this point we can sit back and wait until the proper response arrives.
        String ctag = channel.basicConsume(replyQueueName, true, (consumerTag, delivery) -> {
            if (delivery.getProperties().getCorrelationId().equals(corrId)) {
                response.offer(new String(delivery.getBody(), StandardCharsets.UTF_8));
            }
        }, consumerTag -> {
        });

        /*
         * we're going to need something to suspend the main thread before the response arrives.
         * Usage of BlockingQueue is one possible solutions to do so.Here we are creating ArrayBlockingQueue
         *  with capacity set to 1 as we need to wait for only one response.
         */
        String result = response.take();
        channel.basicCancel(ctag);
        return result;
    }

    @Override
    public void close() throws IOException {
        connection.close();
    }
}
